package com.lhz.manager;

import cn.hutool.core.lang.UUID;
import com.lhz.JobApplicationContext;
import com.lhz.config.ServiceException;
import com.lhz.log.JobLogRecord;
import com.lhz.mapper.JobMapper;
import com.lhz.model.constant.RunnableConstant;
import com.lhz.model.constant.TriggerStatusEnum;
import com.lhz.model.entity.JobInfo;
import com.lhz.utils.CronUtils;
import com.lhz.utils.JobLock;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: LiHuaZhi
 * @Date: 2020/2/29 15:39
 * @Description:
 **/
@Slf4j
public class JobRunnable implements Runnable {

    /**
     * 任务管理
     */
    private JobManager jobManager = JobApplicationContext.getBean(JobManager.class);

    /**
     * 创建任务时，传递任务实体
     */
    private final JobInfo job;

    public JobRunnable(JobInfo job) {
        this.job = job;
    }

    /**
     * 是否进行校验时间差，第一次执行任务时，不校验时间差
     */
    private boolean checkTime = false;

    @Override
    public void run() {

        Long jobId = job.getId();
        String jobName = job.getName();
        String address = RunnableConstant.EXECUTOR_ADDRESS;

        //通过表达式找到需要执行的方法
        String executorHandler = job.getExecutorHandler();
        String executorParam = job.getExecutorParam();

        // 初始化日志记录
        JobLogRecord.init(jobId, jobName, address, executorHandler, executorParam);

        String lockKey = RunnableConstant.JOB_LOCK_KEY + jobId;
        JobLock jobLock = JobApplicationContext.getBean(JobLock.class);
        // redis 锁验证的value
        String value = UUID.fastUUID().toString();

        // 进行redis + lua 分布式锁判断任务判断是否已经执行
        boolean lock = jobLock.setLock(lockKey, value, RunnableConstant.JOB_LOCK_TIME, TimeUnit.MICROSECONDS);

        if (!lock) {
            // 如果已经被锁，则不继续执行，将停止任务的执行
            String msg = "任务:'" + jobName + "' 已执行，暂停当前执行!";
            log.error(msg);
            JobLogRecord.fail(jobId, msg);

            // 任务被锁表示已经在其他机器上执行了，则当前机器进行停止
            jobManager.close(jobId, jobName);
        } else {
            //获取当前执行时间戳
            long currentTime = System.currentTimeMillis();
            try {
                //判断任务执行时间和实际时间差
                LocalDateTime nextRunTime = job.getTriggerNextTime();
                log.info("任务：{}，当前：{}，下次：{}", jobName, LocalDateTime.now(), nextRunTime);
                // 时间差
                long milliSecond = nextRunTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();
                long diffTime = Math.abs(currentTime - milliSecond);

                //执行时，允许误差
                if (diffTime > RunnableConstant.DIFF_TIME && checkTime) {
                    // 获取IP
                    // 抛出异常记录错误日志
                    String sb = "任务执行异常，时间节点错误，相差：" + diffTime + "ms" + ",地址：" + address;
                    throw new ServiceException(sb);
                }

                jobManager.invokeJob(jobId, jobName, executorHandler, executorParam);

                // 解锁
                jobLock.deleteLock(lockKey, value);

                // 清空错误次数记录
                RunnableConstant.FAIL_JOB_NUM.remove(jobId);

            } catch (Exception e) {
                String message = e.getMessage();
                log.error("任务执行失败：{}", message, e);
                JobLogRecord.fail(jobId, message);

                // 任务连续失败三次后将停止运行
                AtomicInteger failNum = RunnableConstant.FAIL_JOB_NUM.get(jobId);
                if (failNum == null) {
                    RunnableConstant.FAIL_JOB_NUM.put(jobId, new AtomicInteger(1));
                } else {
                    if (failNum.intValue() == RunnableConstant.FAIL_MAX) {
                        // 停止任务运行
                        String msg = "连续失败三次，停止任务";

                        log.error(msg);
                        JobLogRecord.fail(jobId, msg);

                        // 停止线程
                        jobManager.stop(job);
                        job.setTriggerStatus(TriggerStatusEnum.STOP.getCode());
                    }

                    failNum.incrementAndGet();
                    RunnableConstant.FAIL_JOB_NUM.put(jobId, failNum);
                }
            } finally {

                // 当任务执行完成后，后续开启时间校验
                checkTime = true;

                // 更新任务
                if (job.getTriggerStatus().equals(TriggerStatusEnum.RUNNING.getCode())) {
                    updateTask(job);
                }

                // 任务结束日志
                JobLogRecord.end(jobId);
            }
        }
    }


    private void updateTask(JobInfo job) {
        // 查询执行周期
        LocalDateTime nextTime = CronUtils.nextCurrentTime(job.getCron());

        //修改任务状况为执行中
        //上次执行时间为，本次的下次执行时间
        job.setTriggerLastTime(job.getTriggerNextTime());
        job.setTriggerNextTime(nextTime);
        //执行中
        job.setTriggerStatus(TriggerStatusEnum.RUNNING.getCode());
        job.setVersion(job.getVersion() + 1);
        JobApplicationContext.getBean(JobMapper.class).updateById(job);

        log.info("更新任务执行情况!");
    }
}
